/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.store.ha;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.apache.rocketmq.store.CommitLog;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.PutMessageSpinLock;
import org.apache.rocketmq.store.PutMessageStatus;

/**
 * 7、主从同步机制
 *
 * 主从broker,对于commitLog文件的同步
 *
 * 流程图: https://kpv4pkozlw.feishu.cn/docx/N5CFdgmVToy25xx16cucFQ2wnig
 *
 *
 * broker主从还有[元数据同步],详见: org.apache.rocketmq.broker.slave.SlaveSynchronize
 *
 *
 *
 * note:RocketMQ主从同步一个重要的特征：主从同步不具备主从切换功能，即当master节点宕机后，slave不会接管消息发送，但可以提供消息读取。
 *      https://www.jianshu.com/p/60e1639161f5
 *
 * note:如果需要使用主从自主切换，需要启用RocketMQ的DLedger模式：
 *         https://zhuanlan.zhihu.com/p/77166786
 *         https://blog.csdn.net/nihui123/article/details/127107500
 */
public class HAService {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    /**
     * client连接数(也就是slave连接数)
     */
    private final AtomicInteger connectionCount = new AtomicInteger(0);

    /**
     * 所有连接的clientSocket的封装类. <br/>
     * 也就是所有的slave broker.
     */
    private final List<HAConnection> connectionList = new LinkedList<>();

    /**
     * 使用javaNIO server socket 监听 HaListenPort,
     * 监听clientSocket的ACCEPT事件.
     * 当有新的clientSocket连接的时候,就把clientSocket封装成 {@link HAConnection},然后添加到 {@link #connectionList}中.
     */
    private final AcceptSocketService acceptSocketService;

    private final DefaultMessageStore defaultMessageStore;

    /**
     * org.apache.rocketmq.store.ha.HAConnection.WriteSocketService#run() -> wait
     *
     */
    private final WaitNotifyObject waitNotifyObject = new WaitNotifyObject();
    /**
     * slave上报的最大偏移量  (commitLog物理偏移量)
     * {@link HAConnection#slaveAckOffset}
     *
     * note:如果一个master broker有2个slave broker,则这个变量表示:2个slave中最大的偏移量.
     */
    private final AtomicLong push2SlaveMaxOffset = new AtomicLong(0);

    private final GroupTransferService groupTransferService;

    private final HAClient haClient;

    public HAService(final DefaultMessageStore defaultMessageStore) throws IOException {
        this.defaultMessageStore = defaultMessageStore;
        // ha监听端口号 (只用于同步commitLog文件的网络I/O)
        int haListenPort = defaultMessageStore.getMessageStoreConfig().getHaListenPort();
        this.acceptSocketService = new AcceptSocketService(haListenPort);

        this.groupTransferService = new GroupTransferService();
        this.haClient = new HAClient();
    }

    public void updateMasterAddress(final String newAddr) {
        if (this.haClient != null) {
            this.haClient.updateMasterAddress(newAddr);
        }
    }

    /**
     * 当有producer生产新消息后,会提交请求,去主从复制.
     * 详见: {@link CommitLog#submitReplicaRequest}
     *
     * @param request
     */
    public void putRequest(final CommitLog.GroupCommitRequest request) {
        this.groupTransferService.putRequest(request);
    }

    public boolean isSlaveOK(final long masterPutWhere) {
        return this.connectionCount.get() > 0 &&
                ((masterPutWhere - this.push2SlaveMaxOffset.get()) < this.defaultMessageStore.getMessageStoreConfig().getHaSlaveFallbehindMax());
    }

    /**
     * 说明:producer发送消息给broker处理时:
     * 1、把消息保存到commitLog
     * 2、提交flush刷盘请求
     * 3、提交replica复制slave请求 —— 如果broker是 SYNC_MASTER 则会 [等待] master和slave之间同步commitLog结束或等待直到超时.
     *
     * 如果slave上报的偏移量是最大的偏移量push2SlaveMaxOffset,说明slave已经把最新的commitLog数据同步完毕,
     * 则会唤醒master broker的 [等待].
     *
     * @param slaveAckOffset slave提交的ack偏移量
     */
    public void notifyTransferSome(final long slaveAckOffset) {
        for (long value = this.push2SlaveMaxOffset.get(); slaveAckOffset > value; ) {
            // 把slave上报的最大偏移量,保存到 push2SlaveMaxOffset 变量中.
            boolean ok = this.push2SlaveMaxOffset.compareAndSet(value, slaveAckOffset);
            if (ok) {
                this.groupTransferService.notifyTransferSome();
                break;
            } else {
                value = this.push2SlaveMaxOffset.get();
            }
        }
    }

    public AtomicInteger getConnectionCount() {
        return connectionCount;
    }

    // public void notifyTransferSome() {
    // this.groupTransferService.notifyTransferSome();
    // }

    public void start() throws Exception {
        // 接收client端socket连接的服务
        this.acceptSocketService.beginAccept();
        this.acceptSocketService.start();// 启动 AcceptSocketService 线程.

        this.groupTransferService.start();// 启动 GroupTransferService 线程
        this.haClient.start(); // 启动 HaClient 线程
    }

    public void addConnection(final HAConnection conn) {
        synchronized (this.connectionList) {
            this.connectionList.add(conn);
        }
    }

    public void removeConnection(final HAConnection conn) {
        synchronized (this.connectionList) {
            this.connectionList.remove(conn);
        }
    }

    public void shutdown() {
        this.haClient.shutdown();
        this.acceptSocketService.shutdown(true);
        this.destroyConnections();
        this.groupTransferService.shutdown();
    }

    public void destroyConnections() {
        synchronized (this.connectionList) {
            for (HAConnection c : this.connectionList) {
                c.shutdown();
            }

            this.connectionList.clear();
        }
    }

    public DefaultMessageStore getDefaultMessageStore() {
        return defaultMessageStore;
    }

    public WaitNotifyObject getWaitNotifyObject() {
        return waitNotifyObject;
    }

    public AtomicLong getPush2SlaveMaxOffset() {
        return push2SlaveMaxOffset;
    }

    /**
     * Listens to slave connections to create {@link HAConnection}.
     *
     * 监听slave的连接并创建 {@link HAConnection}
     */
    class AcceptSocketService extends ServiceThread {
        private final SocketAddress socketAddressListen;
        private ServerSocketChannel serverSocketChannel;
        private Selector selector;

        public AcceptSocketService(final int haListenPort) {
            this.socketAddressListen = new InetSocketAddress(haListenPort);
        }

        /**
         * Starts listening to slave connections.
         *
         * 无论broker主从,都使用nio开启server端.开始监听slave的网络连接.
         * 监听端口号: MessageStoreConfig#getHaListenPort
         */
        public void beginAccept() throws Exception {
            this.serverSocketChannel = ServerSocketChannel.open();
            this.selector = RemotingUtil.openSelector();
            this.serverSocketChannel.socket().setReuseAddress(true);
            // nio server socket 绑定haListenPort
            this.serverSocketChannel.socket().bind(this.socketAddressListen);
            this.serverSocketChannel.configureBlocking(false); // false:非阻塞io
            // select 监听 ACCEPT 客户端连接事件.
            this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public void shutdown(final boolean interrupt) {
            super.shutdown(interrupt);
            try {
                this.serverSocketChannel.close();
                this.selector.close();
            } catch (IOException e) {
                log.error("AcceptSocketService shutdown exception", e);
            }
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public void run() {
            log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    this.selector.select(1000);
                    Set<SelectionKey> selected = this.selector.selectedKeys();

                    if (selected != null) {
                        for (SelectionKey k : selected) {
                            // AcceptSocketService 这个线程主要负责使用javaNIO server socket 监听 haListenPort 端口,监听 ACCEPT 事件.
                            if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
                                // 当有clientSocket连接到这个端口之后,会到这里.获取clientSocket
                                SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();
                                if (sc != null) {
                                    HAService.log.info("HAService receive new connection, " + sc.socket().getRemoteSocketAddress());
                                    try {
                                        HAConnection conn = new HAConnection(HAService.this, sc);
                                        conn.start();
                                        HAService.this.addConnection(conn);
                                    } catch (Exception e) {
                                        log.error("new HAConnection exception", e);
                                        sc.close();
                                    }
                                }
                            } else {
                                log.warn("Unexpected ops in select " + k.readyOps());
                            }
                        }

                        selected.clear();
                    }
                } catch (Exception e) {
                    log.error(this.getServiceName() + " service has exception.", e);
                }
            }

            log.info(this.getServiceName() + " service end");
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public String getServiceName() {
            return AcceptSocketService.class.getSimpleName();
        }
    }

    /**
     * GroupTransferService Service
     *
     * 等待broker主从复制完毕的服务.
     * 作用:
     * 1/ 当有producer生产新消息后,会调用 {@link GroupTransferService#putRequest} 提交一个主从同步的请求.
     * 2/ 异步线程: {@link GroupTransferService#run()}中, '等待'主从数据传输完毕.(重点在'等待',而不是传输.)
     *    如果,主从传输成功,则设置请求的状态为: {@link PutMessageStatus#PUT_OK}
     *    如果,主从传输超时,则设置请求的状态为: {@link PutMessageStatus#FLUSH_SLAVE_TIMEOUT}
     *
     *
     * '等待' 结束有两种情况:
     *   等待1000毫秒之后,线程自己恢复
     *   被唤醒 —— 调用了 {@link GroupTransferService#notifyTransferSome()} 方法.
     */
    class GroupTransferService extends ServiceThread {

        private final WaitNotifyObject notifyTransferObject = new WaitNotifyObject();
        private final PutMessageSpinLock lock = new PutMessageSpinLock();
        /**
         * broker保存消息到commitLog之后,提交的复制请求.
         *
         * {@link CommitLog#submitReplicaRequest}
         */
        private volatile LinkedList<CommitLog.GroupCommitRequest> requestsWrite = new LinkedList<>();
        /**
         * 使用 {@link #swapRequests()} 方法互相转换.
         */
        private volatile LinkedList<CommitLog.GroupCommitRequest> requestsRead = new LinkedList<>();

        /**
         * 当有producer生产新消息后,会提交请求,去主从复制.
         * 详见: {@link CommitLog#submitReplicaRequest}
         *
         * @param request
         */
        public void putRequest(final CommitLog.GroupCommitRequest request) {
            lock.lock();
            try {
                this.requestsWrite.add(request);
            } finally {
                lock.unlock();
            }
            this.wakeup();
        }

        @Override
        public void run() {
            log.info(this.getServiceName() + " service started");
            while (!this.isStopped()) {
                try {
                    // 线程每次执行时,先等待10毫秒.
                    this.waitForRunning(10);
                    // 当等待结束之后,会调用 onWaitEnd() 方法   这个方法中会去调用  swapRequests() 方法.

                    // 等待主从数据传输完毕
                    this.doWaitTransfer();
                } catch (Exception e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }
            log.info(this.getServiceName() + " service end");
        }

        /**
         * 详见 {@link HAService#notifyTransferSome}
         */
        public void notifyTransferSome() {
            this.notifyTransferObject.wakeup();
        }

        private void swapRequests() {
            lock.lock();
            try {
                LinkedList<CommitLog.GroupCommitRequest> tmp = this.requestsWrite;
                this.requestsWrite = this.requestsRead;
                this.requestsRead = tmp;
            } finally {
                lock.unlock();
            }
        }

        /**
         * 等待master传输文件给slave
         *
         * 注意:这里只是等待,不发送数据.
         * 提交的replica请求中:唤醒 WriteSocketService ,是WriteSocketService在发送数据给commitLog.
         * 这里只是等待传输结束,或者等待超时.
         */
        private void doWaitTransfer() {
            if (!this.requestsRead.isEmpty()) {
                for (CommitLog.GroupCommitRequest req : this.requestsRead) {
                    // req.getNextOffset() 表示:producer生产的mq消息保存到commitLog之后的偏移量
                    boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
                    // 等待到何时 (也就是超时时间)
                    long waitUntilWhen = HAService.this.defaultMessageStore.getSystemClock().now() + HAService.this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout();
                    // 在超时时间之前,一直等待同步slave成功
                    while (!transferOK && HAService.this.defaultMessageStore.getSystemClock().now() < waitUntilWhen) {
                        // 备注:slave上报的偏移量slaveAckOffset,如果是最大的,则会唤醒这个等待.   (和push2SlaveMaxOffset相比)
                        this.notifyTransferObject.waitForRunning(1000);
                        transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
                    }

                    if (!transferOK) {
                        // 传输不成功,输出日志.(等待超时了,就不等了)
                        log.warn("transfer messsage to slave timeout, " + req.getNextOffset());
                    }

                    // 设置状态   同步成功, 或 同步超时
                    req.wakeupCustomer(transferOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
                }

                this.requestsRead = new LinkedList<>();
            }
        }

        @Override
        protected void onWaitEnd() {
            // 当线程等待结束后,交换请求的集合.
            this.swapRequests();


            // 什么时候会等待?
            //  run()方法中, 线程每次执行时,先等待10毫秒.  this.waitForRunning(10);
            //  当等待结束之后,会调用 onWaitEnd() 方法. 这个方法中会去调用  swapRequests() 方法.
        }

        @Override
        public String getServiceName() {
            return GroupTransferService.class.getSimpleName();
        }
    }

    /**
     * note:如果当前broker就是master会怎样?
     *  broker的角色是master还是slave都会使用到HAClient
     *
     * 可以把这个类理解为nio的client端.
     * 在broker主从同步时,rpc请求是双向的:
     *  broker可以给slave发消息
     *  slave可以给broker发消息.
     */
    class HAClient extends ServiceThread {
        private static final int READ_MAX_BUFFER_SIZE = 1024 * 1024 * 4;
        /**
         * broker ha master的地址
         * {@link org.apache.rocketmq.broker.BrokerController#getHAServerAddr()}
         *
         * 这个地址,只有slave才会有,
         * 如果broker是master角色,则获取为空.
         * 因为:broker向NameServer注册时:NameServer会做如下处理:
         * <pre>
         *     {@code      org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#registerBroker
         *                 // brokerId不是0,表示当前给NameServer发送请求的是broker slave节点——
         *                 // 则给slave返回master的地址和ha服务地址.
         *                 if (MixAll.MASTER_ID != brokerId) {
         *                     String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
         *                     if (masterAddr != null) {
         *                         // 根据 masterAddr 获取 master的HA地址  (ip相同,端口号不同)
         *                         BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
         *                         if (brokerLiveInfo != null) {
         *                             result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
         *                             result.setMasterAddr(masterAddr);
         *                         }
         *                     }
         *                 }
         *     }
         * </pre>
         */
        private final AtomicReference<String> masterAddress = new AtomicReference<>();
        /**
         * slave给master上报偏移量的buffer.
         * 偏移量是long类型,所以需要8个字节.
         */
        private final ByteBuffer reportOffset = ByteBuffer.allocate(8);
        /**
         * client socket channel
         * 不过这个channel是client主动连接master产生的channel
         * 详见: {@link #connectMaster()}
         */
        private SocketChannel socketChannel;
        /**
         * 关注 OP_READ 事件
         */
        private Selector selector;
        /**
         * 当前需要上报的偏移量
         *
         * 详见: {@link DefaultMessageStore#getMaxPhyOffset}
         */
        private long currentReportedOffset = 0;
        /**
         * 最后一次给master发送数据的时间.
         *
         * 给master发送的数据是: {@link #currentReportedOffset}
         */
        private long lastWriteTimestamp = System.currentTimeMillis();
        /**
         * 已分配的下标.
         * 主要是记录了每次处理 {@link #byteBufferRead} 时的开始下标.
         */
        private int dispatchPosition = 0;
        /**
         * 从master读取到的数据
         */
        private ByteBuffer byteBufferRead = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);
        private ByteBuffer byteBufferBackup = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);

        public HAClient() throws IOException {
            this.selector = RemotingUtil.openSelector();
        }

        @Override
        public void run() {
            while (!this.isStopped()) {
                try {
                    // 1、slave先连接master.如果连接master成功,则继续.
                    if (this.connectMaster()) {

                        // 2、当前时间是否应该给master上报offset
                        if (this.isTimeToReportOffset()) {
                            // 2.1、给master上报最大偏移量
                            boolean result = this.reportSlaveMaxOffset(this.currentReportedOffset);
                            if (!result) {
                                // 上报offset不成功,就关闭master之间的socketChannel网络连接
                                this.closeMaster();
                            }
                        }

                        // 3、等待 OP_READ 事件  (也就是:等待master给slave发送信息.在slave端收到信息,就是信息'可读'事件)
                        //     master给slave发送的消息是:
                        this.selector.select(1000);

                        // 4、处理读事件 —— 处理从master读到的数据.
                        boolean ok = this.processReadEvent();
                        if (!ok) {
                            this.closeMaster();
                        }

                        // 5、如果commitLog偏移量有新增(也就是slave收到新的commitLog数据时),才上报最新的偏移量.
                        if (!reportSlaveMaxOffsetPlus()) {
                            continue;
                        }

                        // 6、判断是否超时.如果超时则断开master的连接.
                        long interval = HAService.this.getDefaultMessageStore().getSystemClock().now() - this.lastWriteTimestamp;
                        if (interval > HAService.this.getDefaultMessageStore().getMessageStoreConfig().getHaHousekeepingInterval()) {
                            this.closeMaster();
                        }
                    } else {
                        // 这里,slave没有成功连接master,则让此线程等5秒.
                        this.waitForRunning(1000 * 5);
                    }
                } catch (Exception e) {
                    this.waitForRunning(1000 * 5);
                }
            }
        }

        /**
         * 修改master地址
         */
        public void updateMasterAddress(final String newAddr) {
            String currentAddr = this.masterAddress.get();
            if (currentAddr == null || !currentAddr.equals(newAddr)) {
                this.masterAddress.set(newAddr);
                log.info("update master address, OLD: " + currentAddr + " NEW: " + newAddr);
            }
        }

        /**
         * 当前时间是否应该给master上报offset
         * <br/>
         * return (now - lastWriteTimestamp) > haSendHeartbeatInterval;
         */
        private boolean isTimeToReportOffset() {
            long interval = HAService.this.defaultMessageStore.getSystemClock().now() - this.lastWriteTimestamp;
            return interval > HAService.this.defaultMessageStore.getMessageStoreConfig().getHaSendHeartbeatInterval();
        }

        /**
         * 上报slave最大的offset
         *
         * @param maxOffset commitLog物理偏移量
         */
        private boolean reportSlaveMaxOffset(final long maxOffset) {
            this.reportOffset.position(0);
            this.reportOffset.limit(8);
            this.reportOffset.putLong(maxOffset);
            this.reportOffset.position(0);
            this.reportOffset.limit(8);

            // 重试发送3次.
            for (int i = 0; i < 3 && this.reportOffset.hasRemaining(); i++) {
                try {
                    this.socketChannel.write(this.reportOffset);
                } catch (IOException e) {
                    log.error(this.getServiceName() + "reportSlaveMaxOffset this.socketChannel.write exception", e);
                    return false;
                }
            }

            // 记录最后一次给master发送数据的时间
            lastWriteTimestamp = HAService.this.defaultMessageStore.getSystemClock().now();
            return !this.reportOffset.hasRemaining();
        }

        private void reallocateByteBuffer() {
            int remain = READ_MAX_BUFFER_SIZE - this.dispatchPosition;
            // byteBufferRead :
            // +-------------------------------+ READ_MAX_BUFFER_SIZE
            // |-------------------|           |   remain 也就是空白位置的长度
            // +-------------------------------+
            //                     dispatchPosition
            if (remain > 0) {
                this.byteBufferRead.position(this.dispatchPosition);

                this.byteBufferBackup.position(0);
                this.byteBufferBackup.limit(READ_MAX_BUFFER_SIZE);
                this.byteBufferBackup.put(this.byteBufferRead);
            }

            // 转换 byteBufferRead 和 byteBufferBackup
            this.swapByteBuffer();

            // 转换之后的 byteBufferRead :
            // +-------------------------------+ READ_MAX_BUFFER_SIZE
            // |    remain|--------------------|
            // +-------------------------------+

            this.byteBufferRead.position(remain);
            this.byteBufferRead.limit(READ_MAX_BUFFER_SIZE);

            // 重置下标.
            this.dispatchPosition = 0;

            // 说明:
            // position=remain; dispatchPosition=0; 在下一次处理的时候,
            // int diff = this.byteBufferRead.position() - this.dispatchPosition;   这里的diff就等于remain
        }

        private void swapByteBuffer() {
            ByteBuffer tmp = this.byteBufferRead;
            this.byteBufferRead = this.byteBufferBackup;
            this.byteBufferBackup = tmp;
        }

        private boolean processReadEvent() {
            // 重试 3次
            int readSizeZeroTimes = 0;
            while (this.byteBufferRead.hasRemaining()) {
                try {
                    // 读取数据
                    int readSize = this.socketChannel.read(this.byteBufferRead);
                    if (readSize > 0) {
                        readSizeZeroTimes = 0;
                        // note:【处理从master读取到的commitLog文件数据.】
                        boolean result = this.dispatchReadRequest();
                        // note:【处理从master读取到的commitLog文件数据.】
                        if (!result) {
                            log.error("HAClient, dispatchReadRequest error");
                            return false;
                        }
                    } else if (readSize == 0) {
                        if (++readSizeZeroTimes >= 3) {
                            break;
                        }
                    } else {
                        log.info("HAClient, processReadEvent read socket < 0");
                        return false;
                    }
                } catch (IOException e) {
                    log.info("HAClient, processReadEvent read socket exception", e);
                    return false;
                }
            }

            return true;
        }

        private boolean dispatchReadRequest() {
            // slave读取到数据头长度
            final int msgHeaderSize = 8 + 4; // phyoffset + size       物理偏移量  +  消息长度

            while (true) {
                int diff = this.byteBufferRead.position() - this.dispatchPosition;
                if (diff >= msgHeaderSize) {
                    // 先读取8字节的偏移量     (master给slave发送的commitLog文件的起始commitLog物理偏移量)
                    long masterPhyOffset = this.byteBufferRead.getLong(this.dispatchPosition);
                    // 数据包体长度.4字节
                    int bodySize = this.byteBufferRead.getInt(this.dispatchPosition + 8);
                    // slave当前最大物理偏移量
                    long slavePhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();

                    if (slavePhyOffset != 0) {
                        // 如果slave的偏移量和master的偏移量不一样,说明此次发来的数据包可能是slave已经处理过的 或者 消息不连续的.则返回错误.
                        if (slavePhyOffset != masterPhyOffset) {
                            log.error("master pushed offset not equal the max phy offset in slave, SLAVE: " + slavePhyOffset + " MASTER: " + masterPhyOffset);
                            return false;
                        }
                    }

                    // 最新读取到的数据长度,大于,消息头+body的长度,表示至少有一条完整的消息
                    //     (如果不成立,说明消息不完整,则继续使用从socketChannel读取数据,直到一条数据完整读取.)
                    if (diff >= (msgHeaderSize + bodySize)) {
                        byte[] bodyData = byteBufferRead.array(); // 把所有读取到byte字节流都获取到.
                        int dataStart = this.dispatchPosition + msgHeaderSize;

                        // 追加commitLog数据.  这里只处理一条数据.也就是 长度= bodySize.   bodySize之后的数据会在continue等待下一次遍历时处理.
                        HAService.this.defaultMessageStore.appendToCommitLog(masterPhyOffset, bodyData, dataStart, bodySize);

                        // 设置已经读取数据位置
                        this.dispatchPosition += msgHeaderSize + bodySize;

                        // 如果commitLog文件有增加,则给master上报偏移量
                        if (!reportSlaveMaxOffsetPlus()) {
                            return false;
                        }

                        // 再次循环,主要目的是:处理 byteBufferRead中 index = (msgHeaderSize + bodySize)  之后的数据.
                        continue;
                    }
                } // diff >= msgHeaderSize 结束

                // 到这里,说明读取到的数据,已经不是完整的  header + body 了.

                //  byteBufferRead中的数据都处理完毕了,没有剩余
                //      (也就是byteBufferRead中的数据,正好是 header + body 的一条完整的数据,并且没有信的数据收到了.
                if (!this.byteBufferRead.hasRemaining()) {
                    // byteBufferRead和byteBufferBackup转换
                    this.reallocateByteBuffer();
                }

                break;
            } // while (true)  结束

            return true;
        }

        /**
         * 和 {@link #reportSlaveMaxOffset(long)} 方法的区别:
         * 如果commitLog最大物理偏移量比上一次上报的偏移量大,才会上报.否则不上报.(也就是有新数据从master拉取过来了,才会上报偏移量)
         *
         * @return 是否成功
         */
        private boolean reportSlaveMaxOffsetPlus() {
            boolean result = true;
            // (slave的)commitLog最大物理偏移量
            long currentPhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
            // 如果commitLog最大物理偏移量比上一次上报的偏移量大,才会上报.否则忽略.
            if (currentPhyOffset > this.currentReportedOffset) {
                this.currentReportedOffset = currentPhyOffset;
                result = this.reportSlaveMaxOffset(this.currentReportedOffset);
                if (!result) {
                    this.closeMaster();
                    log.error("HAClient, reportSlaveMaxOffset error, " + this.currentReportedOffset);
                }
            }

            return result;
        }

        /**
         * 连接master节点
         */
        private boolean connectMaster() throws ClosedChannelException {
            if (null == socketChannel) {// 为null,说明当前slave没有连接master
                // master节点的地址.    (masterAddress如果为空,说明没有master的地址  , RegisterBrokerResult#haServerAddr )
                String addr = this.masterAddress.get();
                if (addr != null) {
                    SocketAddress socketAddress = RemotingUtil.string2SocketAddress(addr);
                    if (socketAddress != null) {
                        this.socketChannel = RemotingUtil.connect(socketAddress);
                        if (this.socketChannel != null) {
                            this.socketChannel.register(this.selector, SelectionKey.OP_READ);
                        }
                    }
                }

                // 从store存储模块中,获取最大的物理偏移量.
                this.currentReportedOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();

                this.lastWriteTimestamp = System.currentTimeMillis();
            }

            // 如果有socket,则说明连接master成功.
            return this.socketChannel != null;
        }

        private void closeMaster() {
            if (null != this.socketChannel) {
                try {

                    SelectionKey sk = this.socketChannel.keyFor(this.selector);
                    if (sk != null) {
                        sk.cancel();
                    }

                    this.socketChannel.close();

                    this.socketChannel = null;
                } catch (IOException e) {
                    log.warn("closeMaster exception. ", e);
                }

                this.lastWriteTimestamp = 0;
                this.dispatchPosition = 0;

                this.byteBufferBackup.position(0);
                this.byteBufferBackup.limit(READ_MAX_BUFFER_SIZE);

                this.byteBufferRead.position(0);
                this.byteBufferRead.limit(READ_MAX_BUFFER_SIZE);
            }
        }

        @Override
        public void shutdown() {
            super.shutdown();
            closeMaster();
        }

        // private void disableWriteFlag() {
        // if (this.socketChannel != null) {
        // SelectionKey sk = this.socketChannel.keyFor(this.selector);
        // if (sk != null) {
        // int ops = sk.interestOps();
        // ops &= ~SelectionKey.OP_WRITE;
        // sk.interestOps(ops);
        // }
        // }
        // }
        // private void enableWriteFlag() {
        // if (this.socketChannel != null) {
        // SelectionKey sk = this.socketChannel.keyFor(this.selector);
        // if (sk != null) {
        // int ops = sk.interestOps();
        // ops |= SelectionKey.OP_WRITE;
        // sk.interestOps(ops);
        // }
        // }
        // }

        @Override
        public String getServiceName() {
            return HAClient.class.getSimpleName();
        }
    }
}
